package com.源码分析.processor;

import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;
import reactor.util.concurrent.Queues;

// 支持backpressure特性，但是代价是至多只能有一个subscriber。默认是无界的，如果发布数据之后，subscriber还没来得及request，则它会把数据缓存下来。
// 如果设置了一个有界的queue，当buffer满而且subscriber没有发送足够多的request的时候，processor会拒绝推送数据。在这种场景下，processor内置了一个callback，每当一个element被rejected的时候会触发.
public class UnicastProcessorDemo {
	static final Logger LOGGER = LoggerFactory.getLogger(UnicastProcessorDemo.class);

	@Test
    public void testUnicastProcessor() throws InterruptedException {
        UnicastProcessor<Integer> unicastProcessor = UnicastProcessor.create(Queues.<Integer>get(8).get());
        Flux<Integer> flux = unicastProcessor
                .map(e -> e);
        //UnicastProcessor allows only a single Subscriber
        flux.subscribe(e -> {
            LOGGER.info("flux subscriber:{}",e);
        });

        IntStream.rangeClosed(1,12)
                .forEach(e -> {
                    LOGGER.info("emit:{}",e);
                    unicastProcessor.onNext(e);
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
        LOGGER.info("begin to sleep 7 seconds");
        TimeUnit.SECONDS.sleep(7);

        unicastProcessor.onComplete();
        TimeUnit.SECONDS.sleep(10);
//        unicastProcessor.blockLast(); //blockLast也是一个subscriber
    }

}
